package hk.linkcredit.walletservice.wallet.consumer;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import hk.linkcredit.walletservice.constant.Status;
import hk.linkcredit.walletservice.constant.WalletConstants;
import hk.linkcredit.walletservice.exception.BizException;
import hk.linkcredit.walletservice.utils.core.RedisCache;
import hk.linkcredit.walletservice.wallet.domain.entity.WalletFlow;
import hk.linkcredit.walletservice.wallet.domain.vo.WalletAccountReq;
import hk.linkcredit.walletservice.wallet.service.WalletAccountService;
import hk.linkcredit.walletservice.wallet.service.WalletFlowService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

/**
 * 钱包账户更新消费
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.consumers.walletAccountConsumerGroup}", topic = "${rocketmq.topics.walletAccountTopic}",
        messageModel = MessageModel.CLUSTERING,
        consumeMode = ConsumeMode.CONCURRENTLY)
public class WalletAccountConsumer implements RocketMQListener<MessageExt> {
    @Autowired
    private WalletAccountService walletAccountService;
    @Autowired
    private WalletFlowService walletFlowService;

    @Autowired
    RedisCache redisCache;

    /**
     * @param message [{ "reference":"290","custNo":"20220527636202569730", "amount":10.00, "fundsType":"1" }]
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void onMessage(MessageExt message) {
        long start = System.currentTimeMillis();
        log.info("WalletAccountConsumer--> message：{}", JSONObject.toJSONString(message));

        //业务防重
        String msgId = message.getMsgId();
        List<WalletFlow> walletFlowList = walletFlowService.list(new LambdaQueryWrapper<WalletFlow>().eq(WalletFlow::getMsgId, msgId));
        if (CollectionUtils.isNotEmpty(walletFlowList)) {
            log.info("重复消费直接提交");
            return;
        }

        List<WalletAccountReq> walletAccountReqList = JSONArray.parseArray(new String(message.getBody()), WalletAccountReq.class);
        walletAccountReqList.forEach(obj -> {
            obj.setMsgId(msgId);
            updateWalletAccountAmount(obj);
        });

        log.info("WalletAccountConsumer-->handler message：{} ms", System.currentTimeMillis() - start);
    }

    private void updateWalletAccountAmount(WalletAccountReq obj) {
        String lockKey = WalletConstants.WALLET_ACCOUNT_LOCK_KEY_PREFIX + obj.getCustNo();
        RLock lock = redisCache.lock(lockKey);
        try {
            long lockStart = System.currentTimeMillis();
            walletAccountService.updateWalletAccount(obj);
            log.info("锁占用时间，耗时：{} ms [ {} ]", (System.currentTimeMillis() - lockStart), lockKey);
        } catch (Exception e) {
            log.error("更新钱包账户异常：", e);
            throw new BizException(Status.WALLE_OPERATE_ERROR);
        } finally {
            lock.unlock();
        }
    }
}